package pro.coman404.tools;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.StreamFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import pro.coman404.tools.pojo.Person;

import javax.xml.crypto.Data;
import java.io.File;
import java.util.Arrays;
import java.util.List;

/**
 * @author NotFound404 at 2024/7/2
 * code code always code
 */
public class DataStreamApi {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        Tuple2<String,Integer> persion = Tuple2.of("coman", 18);
//        DataStream<Tuple2<String,Integer>> dataStream =  env.fromData(persion);

//        DataStream<Person> dataStream = env.fromData(
//                new Person("Fraed", 35),
//                new Person("Wilma", 18),
//                new Person("coman", 2)
//        );

//        DataStream<String> lines = env.socketTextStream("localhost",8089);

//        StreamFormat<String> format = new TextLineInputFormat();
//        DataStream<String> lines = env.fromSource(FileSource.forRecordStreamFormat(format,Path.fromLocalFile(new File("demo.txt"))).build(),
//                WatermarkStrategy.noWatermarks(), "file");
        List<Person> personList = Arrays.asList(new Person("Fraed", 35),
                new Person("Wilma", 18),
                new Person("coman", 2));

        DataStream<Person> dataStream = env.fromData(personList);
        DataStream<Person> adults = dataStream.filter(person -> {
            return person.getAge()>=18;
        });

        adults.print();

        env.execute();

    }

}
